package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/3/14
 */
public class App12_DwdMhsIptFirstCodRecord extends BaseSqlApp {
    public static void main(String[] args) {
        new App12_DwdMhsIptFirstCodRecord().init(
                5012,
                "App12_DwdMhsIptFirstCodRecord",
                2
        );
    }

    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        // create kafka mappings
        tableEnv.executeSql("CREATE TABLE target (      "+
                "    RECORD_ID                         STRING    "+
                "    ,HOSPITAL_NAME                    STRING    "+
                "    ,HOSPITAL_CODE                    STRING    "+
                "    ,AGE_UNIT                         STRING    "+
                "    ,AGE                              STRING    "+
                "    ,SEX_NAME                         STRING    "+
                "    ,SEX_CODE                         STRING    "+
                "    ,ID_CARD_NUMBER                   STRING    "+
                "    ,PATIENT_VISIT_CATEGORY_CODE      STRING    "+
                "    ,PATIENT_VISIT_CATEGORY           STRING    "+
                "    ,INPATIENT_VISIT_FLOW_ID          STRING    "+
                "    ,INPATIENT_NO                     STRING    "+
                "    ,PATIENT_ID                       STRING    "+
                "    ,PATIENT_NAME                     STRING    "+
                "    ,DOMAIN_ID                        STRING    "+
                "    ,VISIT_TIMES                      STRING    "+
                "    ,BIRTH_DATE                       STRING    "+
                "    ,AUTHOR_CODE                      STRING    "+
                "    ,AUTHOR_NAME                      STRING    "+
                "    ,CUSTODIAN_CODE                   STRING    "+
                "    ,CUSTODIAN_NAME                   STRING    "+
                "    ,SENIOR_DOCTORE_SIGN_TIME         STRING    "+
                "    ,SENIOR_DOCTOR_ID                 STRING    "+
                "    ,SENIOR_DOCTOR_SIGN_NAME          STRING    "+
                "    ,RESIDENT_DOCTORE_SIGN_TIME       STRING    "+
                "    ,RESIDENT_DOCTOR_ID               STRING    "+
                "    ,RESIDENT_DOCTOR_SIGN_NAME        STRING    "+
                "    ,VISIT_DATE                       STRING    "+
                "    ,BED_NO                           STRING    "+
                "    ,BED_NAME                         STRING    "+
                "    ,ROOM_NO                          STRING    "+
                "    ,ROOM_NAME                        STRING    "+
                "    ,WARD_ID                          STRING    "+
                "    ,WARD_NAME                        STRING    "+
                "    ,DEPT_ID                          STRING    "+
                "    ,DEPT_NAME                        STRING    "+
                "    ,CHIEF_COMPLAINT                  STRING    "+
                "    ,CASE_CHARACTER                   STRING    "+
                "    ,CHINA_MED_OBSERVE_RESULT         STRING    "+
                "    ,DIAG_BASIS                       STRING    "+
                "    ,TREAT_PLAN                       STRING    "+
                "    ,THERAPEUTIC_PRINCIPLE            STRING    "+
                "    ,PREL_DIAG_ICD_CODE               STRING    "+
                "    ,PREL_DIAG_ICD_NAME               STRING    "+
                "    ,PREL_CHINA_DISEASE_CODE          STRING    "+
                "    ,PREL_CHINA_DISEASE_NAME          STRING    "+
                "    ,PREL_CHINA_SYNDROME_CODE         STRING    "+
                "    ,PREL_CHINA_SYNDROME_NAME         STRING    "+
                "    ,DIFF_DIAG_ICD_CODE               STRING    "+
                "    ,DIFF_DIAG_ICD_NAME               STRING    "+
                "    ,DIFF_CHINA_DISEASE_CODE          STRING    "+
                "    ,DIFF_CHINA_DISEASE_NAME          STRING    "+
                "    ,DIFF_CHINA_SYNDROME_CODE         STRING    "+
                "    ,DIFF_CHINA_SYNDROME_NAME         STRING    " +
                "    ,VISIT_ID                         STRING    "+
                "    ,PRIMARY KEY (RECORD_ID) NOT ENFORCED       "+
                ") "+ FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_IPT_FIRST_COD_RECORD));
        // load data
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        // 1 设置数据的TTL（暂时10min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        // 2 join操作
        Table joinResult = tableEnv.sqlQuery("SELECT a.unique_id    AS    RECORD_ID                                   "+
                "     ,a.hospital_name            AS     HOSPITAL_NAME                                                                   "+
                "     ,a.hospital_code            AS     HOSPITAL_CODE                                                                   "+
                "     ,a.age_unit                 AS     AGE_UNIT                                                                        "+
                "     ,a.age                      AS     AGE                                                                             "+
                "     ,a.sex                      AS     SEX_NAME                                                                        "+
                "     ,a.sex_code                 AS     SEX_CODE                                                                        "+
                "     ,a.identify_no              AS     ID_CARD_NUMBER                                                                  "+
                "     ,a.patient_typecode         AS     PATIENT_VISIT_CATEGORY_CODE                                                     "+
                "     ,a.patient_type             AS     PATIENT_VISIT_CATEGORY                                                          "+
                "     ,a.visit_id                 AS     INPATIENT_VISIT_FLOW_ID                                                         "+
                "     ,a.inpatient                AS     INPATIENT_NO                                                                    "+
                "     ,a.patient_id               AS     PATIENT_ID                                                                      "+
                "     ,a.patient_name             AS     PATIENT_NAME                                                                    "+
                "     ,a.domain_id                AS     DOMAIN_ID                                                                       "+
                "     ,a.visit_times              AS     VISIT_TIMES                                                                     "+
                "     ,a.birth                    AS     BIRTH_DATE                                                                      "+
                "     ,a.author_code              AS     AUTHOR_CODE                                                                     "+
                "     ,a.author_name              AS     AUTHOR_NAME                                                                     "+
                "     ,a.custodian_code           AS     CUSTODIAN_CODE                                                                  "+
                "     ,a.custodian_name           AS     CUSTODIAN_NAME                                                                  "+
                "     ,a.superior_signdate        AS     SENIOR_DOCTORE_SIGN_TIME                                                        "+
                "     ,a.superior_code            AS     SENIOR_DOCTOR_ID                                                                "+
                "     ,a.superior_name            AS     SENIOR_DOCTOR_SIGN_NAME                                                         "+
                "     ,a.resident_signdate        AS     RESIDENT_DOCTORE_SIGN_TIME                                                      "+
                "     ,a.resident_code            AS     RESIDENT_DOCTOR_ID                                                              "+
                "     ,a.resident_name            AS     RESIDENT_DOCTOR_SIGN_NAME                                                       "+
                "     ,a.visit_date               AS     VISIT_DATE                                                                      "+
                "     ,a.bed_no                   AS     BED_NO                                                                          "+
                "     ,a.bed_name                 AS     BED_NAME                                                                        "+
                "     ,a.ward_id                  AS     ROOM_NO                                                                         "+
                "     ,a.ward_name                AS     ROOM_NAME                                                                       "+
                "     ,a.wards_id                 AS     WARD_ID                                                                         "+
                "     ,a.wards_name               AS     WARD_NAME                                                                       "+
                "     ,a.dept_code                AS     DEPT_ID                                                                         "+
                "     ,a.dept_name                AS     DEPT_NAME                                                                       "+
                "     ,a.chief_complaint          AS     CHIEF_COMPLAINT                                                                 "+
                "     ,a.case_characteristics     AS     CASE_CHARACTER                                                                  "+
                "     ,a.tcm_four_diagnosis       AS     CHINA_MED_OBSERVE_RESULT                                                        "+
                "     ,a.diagnostic_basis         AS     DIAG_BASIS                                                                      "+
                "     ,a.assessment_plan          AS     TREAT_PLAN                                                                      "+
                "     ,a.principle_and_method     AS     THERAPEUTIC_PRINCIPLE                                                           "+
                "     ,case when b.diag_type='初步诊断-西医诊断' then b.diag_code else '' end    AS    PREL_DIAG_ICD_CODE                "+
                "     ,case when b.diag_type='初步诊断-西医诊断' then b.diag_name else '' end    AS    PREL_DIAG_ICD_NAME                "+
                "     ,case when b.diag_type='初步诊断-中医病名' then b.diag_code else '' end    AS    PREL_CHINA_DISEASE_CODE           "+
                "     ,case when b.diag_type='初步诊断-中医病名' then b.diag_name else '' end    AS    PREL_CHINA_DISEASE_NAME           "+
                "     ,case when b.diag_type='初步诊断-中医证候' then b.diag_code else '' end    AS    PREL_CHINA_SYNDROME_CODE          "+
                "     ,case when b.diag_type='初步诊断-中医证候' then b.diag_name else '' end    AS    PREL_CHINA_SYNDROME_NAME          "+
                "     ,case when b.diag_type='鉴别诊断-西医诊断' then b.diag_code else '' end    AS    DIFF_DIAG_ICD_CODE                "+
                "     ,case when b.diag_type='鉴别诊断-西医诊断' then b.diag_name else '' end    AS    DIFF_DIAG_ICD_NAME                "+
                "     ,case when b.diag_type='鉴别诊断-中医病名' then b.diag_code else '' end    AS    DIFF_CHINA_DISEASE_CODE           "+
                "     ,case when b.diag_type='鉴别诊断-中医病名' then b.diag_name else '' end    AS    DIFF_CHINA_DISEASE_NAME           "+
                "     ,case when b.diag_type='鉴别诊断-中医证候' then b.diag_code else '' end    AS    DIFF_CHINA_SYNDROME_CODE          "+
                "     ,case when b.diag_type='鉴别诊断-中医证候' then b.diag_name else '' end    AS    DIFF_CHINA_SYNDROME_NAME          " +
                "     ,visit_id AS VISIT_ID "+
                " FROM hdsd00_14_14 a                                                                                                    "+
                " LEFT JOIN tdy_list_diag b                                                                                              "+
                " on a.unique_id = b.unique_id and a.xds_type = b.xds_type                                                             ");
        tableEnv.createTemporaryView("join_result",joinResult);

    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        // create table : hdsd00_14_14
        tableEnv.executeSql("create table hdsd00_14_14 (       " +
                "pk                       string," +
                "upload_time              string," +
                "status                   string," +
                "empi                     string," +
                "encounter_id             string," +
                "patient_id               string," +
                "patient_domain           string," +
                "visit_domain             string," +
                "visit_id                 string," +
                "visit_times              string," +
                "unique_id                string," +
                "xds_type                 string," +
                "xds_name                 string," +
                "effective_time           string," +
                "xds_version              string," +
                "domain_id                string," +
                "patient_type             string," +
                "patient_typecode         string," +
                "inpatient                string," +
                "identify_no              string," +
                "patient_name             string," +
                "sex_code                 string," +
                "sex                      string," +
                "birth                    string," +
                "age                      string," +
                "age_unit                 string," +
                "write_time               string," +
                "author_code              string," +
                "author_name              string," +
                "custodian_code           string," +
                "custodian_name           string," +
                "superior_signdate        string," +
                "superior_code            string," +
                "superior_name            string," +
                "resident_signdate        string," +
                "resident_code            string," +
                "resident_name            string," +
                "visit_date               string," +
                "bed_no                   string," +
                "bed_name                 string," +
                "ward_id                  string," +
                "ward_name                string," +
                "wards_id                 string," +
                "wards_name               string," +
                "dept_code                string," +
                "dept_name                string," +
                "hospital_code            string," +
                "hospital_name            string," +
                "chief_complaint          string," +
                "case_characteristics     string," +
                "tcm_four_diagnosis       string," +
                "diagnostic_basis         string," +
                "assessment_plan          string," +
                "principle_and_method     string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_14_14", "App12_DwdMhsIptFirstCodRecord"));

        // create table : tdy_list_diag
        super.readTdyListDiag(tableEnv, "App12_DwdMhsIptFirstCodRecord");
    }
}
